package org.yantsu.loadtestanalyzer.ltaweb.importer;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.NumberFormat;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Locale;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.Map.Entry;
import java.util.zip.GZIPInputStream;

import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.yantsu.loadtestanalyzer.ltaweb.ElasticSearchClient;

@Component
public class ImportScanner {
	private static final Logger LOG = Logger.getLogger(ImportScanner.class);
	private static final String IMPORTDIR_PROP = "org.yantsu.lta.importpath";
	private static final int BULK_SIZE = 4096; 
	
	private NumberFormat valueFormat = NumberFormat.getInstance(Locale.GERMAN);
	private File importDir;
	
	public static class Data {
		public long timestamp = -1;
		public String hostname;
		public String category;
		public Map<String, Double> values = new HashMap<String, Double>();
		public Map<String, String> properties = new HashMap<String, String>();
	}

	@Autowired
	public ElasticSearchClient client;
	
	public ImportScanner() {
		String importDirStr = System.getProperty(IMPORTDIR_PROP);
		if(importDirStr == null) {
			throw new IllegalArgumentException("Missing configuration property: " + IMPORTDIR_PROP);
		}
		importDir = new File(importDirStr);
		if(!importDir.exists() || !importDir.isDirectory()) {
			throw new IllegalArgumentException("Import directory " + IMPORTDIR_PROP + " does not exist or is not a directoy");
		}
	}
	
	@Scheduled(fixedRate=30000)
	public void scanImportDir() {
		if(LOG.isInfoEnabled()) {
			LOG.info("Scanning the import directory " + importDir.getAbsolutePath());
		}
		File[] scenarios = getScenarios(importDir);
		for (File scenario : scenarios) {
			LOG.info("Scanning scenario " + scenario.getAbsolutePath());
			File[] machines = getMachines(scenario);
			for (File machine : machines) {
				LOG.info("Scanning machine " + machine.getAbsolutePath());
				processGrinderData(scenario.getName(), machine);
				processSarData(scenario.getName(), machine);
			}
		}
	}
	
	private File[] getScenarios(File importDir) {
		return importDir.listFiles(new FileFilter() {
			public boolean accept(File pathname) {
				return pathname.isDirectory();
			}
		});
	}

	private File[] getMachines(File scenarioDir) {
		return scenarioDir.listFiles(new FileFilter() {
			public boolean accept(File pathname) {
				return pathname.isDirectory();
			}
		});
	}
	
	private File getGrinderData(File machineDir) {
		File grinderDir = new File(machineDir, "grinder");
		if(grinderDir.exists() && grinderDir.isDirectory()) {
			File compressedData = new File(grinderDir, machineDir.getName() + "-0-data.log.gz");
			if (compressedData.exists() && compressedData.isFile()) return compressedData;
			File uncompressedData = new File(grinderDir, machineDir.getName() + "-0-data.log");
			if (uncompressedData.exists() && uncompressedData.isFile()) return uncompressedData;
		}
		return null;
	}
	
	private void processGrinderData(String testname, File machineDir) {
		File data = getGrinderData(machineDir);
		if(data != null) {
			long startTime = System.currentTimeMillis();
			if(LOG.isInfoEnabled()) {
				LOG.info("Processing grinder data for " + machineDir.getName());
			}
			BufferedReader br = null;
			int lineNo = 1;
			long start = -1;
			long end = -1;
			try {
				InputStream is = new FileInputStream(data);
				if(data.getName().endsWith(".gz")) {
					is = new GZIPInputStream(is);
				}
				br = new BufferedReader(new InputStreamReader(is));
				String line = br.readLine();
				if(line != null) line = br.readLine(); //ignore 1st line with header
				lineNo++;

				int bulkCount = 0;
				BulkRequestBuilder bulkRequest = client.prepareBulk();
				String id = client.getSeriesDocAndLock("Grinder", testname);
				
				while(line != null) {
					StringTokenizer tokenizer = new StringTokenizer(line, ", ");
					if(tokenizer.countTokens() != 12) {
						throw new IllegalStateException("Format error in Grinder data file in line " + lineNo + ": " + line);
					}

					XContentBuilder builder = jsonBuilder();
					builder.startObject();
					builder.field("hostname", "N/A");	
					
					tokenizer.nextToken(); //thread
					tokenizer.nextToken(); //run

					builder.field("category", tokenizer.nextToken());	
					
					//validation and typing
					String timestamp = tokenizer.nextToken();
					long timestampL = -1;
					try {
						timestampL = Long.valueOf(timestamp);
					} catch(NumberFormatException e) {
						throw new IllegalStateException("Timestamp retrieved in line "+ lineNo + " is not a number, was " + timestamp, e);
					}
					builder.field("timestamp", timestampL);
					
					if(start == -1) {
						start = timestampL;
					}
					
					builder.startObject("values");
					
					String name = "";
					try {
						name = "testtime";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "errors";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "responsecode";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "responselength";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "responseerrors";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "ttrh";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "ttec";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
						name = "ttfb";
						builder.field(name, valueFormat.parse(tokenizer.nextToken()).doubleValue());
					} catch(NumberFormatException e) {
						throw new IllegalStateException(name + " retrieved in line " + lineNo + " is not a number", e);
					} catch (ParseException e) {
						throw new IllegalStateException(name + " retrieved in line " + lineNo + " is not a number", e);
					}					
					builder.endObject();
					builder.endObject();
					builder.close();

					bulkRequest.add(client.prepareData("Grinder", id, builder));
					if(++bulkCount > BULK_SIZE) {
						client.writeBulk(bulkRequest);
						bulkCount = 0;
						bulkRequest = client.prepareBulk();
					}
					
					end = timestampL;
					line = br.readLine();
					lineNo++;
				}
				if(bulkCount != 0) {
					client.writeBulk(bulkRequest);
				}		
				if(!data.delete()) {
					LOG.error("Imported file cannot be deleted: " + data.getAbsolutePath());
				}
			} catch (FileNotFoundException e) {
				LOG.error("Grinder data file not found: " + data.getAbsolutePath());
			} catch (IOException e) {
				LOG.error("Error reading data file");
			} finally {
				if(br != null) {
					try {
						br.close();
					} catch (IOException e) {}
					client.finalizeSeriesDoc(testname, "Grinder", start, end);
				}
			}
			if(LOG.isInfoEnabled()) {
				LOG.info("Grinder data imported in " + (System.currentTimeMillis() - startTime) + "ms");
			}
		}
	}
	
	private void processSarData(String testname, File machineDir) {
		File dataFile = null;
		BufferedReader br = null;
		int lineNo = 1;
		long start = -1;
		long end = -1;
		
		Data data = null;
		Data nextData = new Data();
		String cat;
		
		try {
			InputStream is = null;
			dataFile = new File(machineDir, "sysstat.txt.gz");
			if(dataFile.exists() && dataFile.isFile()) {
				is = new GZIPInputStream(new FileInputStream(dataFile));
			} else {
				dataFile = new File(machineDir, "sysstat.txt");
				if(dataFile.exists() && dataFile.isFile()) {
					is = new FileInputStream(dataFile);
				}
			}
			if(is != null) {
				long startTime = System.currentTimeMillis();
				
				br = new BufferedReader(new InputStreamReader(is));
				String line = br.readLine();
				lineNo++;

				int bulkCount = 0;
				BulkRequestBuilder bulkRequest = client.prepareBulk();
				String id = client.getSeriesDocAndLock("Sar", testname);
				
				while(line != null) {
					data = nextData;
					
					StringTokenizer tokenizer = new StringTokenizer(line, "\t");
					if(tokenizer.countTokens() != 6) {
						throw new IllegalStateException("Sar format error: Token count in line " + lineNo + " is not 6");
					}
					
					String hostname = tokenizer.nextToken();
					tokenizer.nextToken(); //interval
					String timestamp = tokenizer.nextToken();
					String category = tokenizer.nextToken();
					String type = tokenizer.nextToken();
					String value = tokenizer.nextToken();
					
					//validation and typing
					long timestampL = -1;
					try {
						timestampL = Long.valueOf(timestamp);
					} catch(NumberFormatException e) {
						throw new IllegalStateException("Timestamp retrieved in line "+ lineNo + " is not a number, was " + timestamp, e);
					}

					if(start == -1) {
						start = timestampL;
					}

					double valueD = 0.0;
					try {
						NumberFormat nf = NumberFormat.getInstance(Locale.GERMAN);
						valueD = nf.parse(value).doubleValue();
					} catch(NumberFormatException e) {
						throw new IllegalStateException("Value retrieved in line " + lineNo + " is not a number, was " + value, e);
					} catch (ParseException e) {
						throw new IllegalStateException("Value retrieved in line " + lineNo + " is not a number, was " + value, e);
					}
					timestampL = timestampL * 1000L;
	
					if(timestampL == -1) {
						throw new IllegalStateException("Timestamp in line " + lineNo + " is invalid, was " + timestamp);
					}
					
					if(data.timestamp != -1 && (data.timestamp != timestampL || !"-".equals(category) && !data.category.equals(category))) { //new group starting
						nextData = new Data();
						nextData.timestamp = timestampL;
						nextData.hostname = hostname; 
						nextData.values.put(type, valueD);
						if("-".equals(category)) {
							cat = type;
							nextData.category = cat;
						} else {
							cat = null;
							nextData.category = category;
						}

						XContentBuilder builder = jsonBuilder();
						builder.startObject();
						builder.field("timestamp", data.timestamp);
						builder.field("hostname", data.hostname);	
						builder.field("category", data.category);	
						builder.startObject("values");
						for (Iterator<Entry<String, Double>> i = data.values.entrySet().iterator(); i.hasNext();) {
							Entry<String, Double> entry = i.next();
							builder.field(entry.getKey(), entry.getValue());
						}
						builder.endObject();
						builder.endObject();
						builder.close();
						
						bulkRequest.add(client.prepareData("Sar", id, builder));
						if(++bulkCount > BULK_SIZE) {
							client.writeBulk(bulkRequest);
							bulkCount = 0;
							bulkRequest = client.prepareBulk();
						}
						
					} else {
						if(data.timestamp == -1) { //the very first data
							data.timestamp = timestampL;
							data.hostname = hostname;
							if("-".equals(category)) {
								cat = type;
								data.category = cat;
							} else {
								cat = null;
								data.category = category;
							}
						} else if(!data.hostname.equals(hostname)) {
							throw new IllegalStateException("Hostname has changed inside a group with same timestamps. Line " + lineNo);
						}
						data.values.put(type, valueD);
					}
										
					end = timestampL;
					line = br.readLine();
					lineNo++;
				}
				if(bulkCount != 0) {
					client.writeBulk(bulkRequest);
				}		
				if(!dataFile.delete()) {
					LOG.error("Imported file cannot be deleted: " + dataFile.getAbsolutePath());
				}
				
				if(LOG.isInfoEnabled()) {
					LOG.info("Sar data imported in " + (System.currentTimeMillis() - startTime) + "ms");
				}				
			}
		} catch (FileNotFoundException e) {
			LOG.error("Sar data file not found: " + dataFile.getAbsolutePath());
		} catch (IOException e) {
			LOG.error("Error reading data file");
		} finally {
			if(br != null) {
				try {
					br.close();
				} catch (IOException e) {}
				client.finalizeSeriesDoc(testname, "Sar", start, end);
			}
		}
	}
}
